//Copyright © 2012 Ryan Larson, Ben Glasser
//[This program is licensed under the "GPL v3.0 License"]
// Please see the file COPYING in the source
//distribution of this software for license terms.
package org.app_chat.activemq;

import com.google.common.base.Throwables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.protobuf.Message;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.app_chat.api.Listener;
import org.app_chat.api.Publisher;
import org.app_chat.service.protobuf.MessageUnMarshaller;

import javax.jms.*;
import java.util.Map;
import java.util.Set;

/*
 * Created by Ryan of app-chat.org
 * on: 7/12/12 at 8:17 PM
 */

/**
 * ActiveMQ Implementation of the ChatRoom
 */

public class ChatRoom {

    private final String brokerUrl;
    private final Set<Listener> subscriptions;

    private final Map<String, Destination> topicNamesToTopics = Maps.newHashMap();
    private final Map<String, MessageProducer> queueNamesToProducers = Maps.newHashMap();

    private Session session;
    private PublisherImpl publisher;

    private ConnectionFactory connectionFactory;

    private ChatRoom(Builder builder) {
        subscriptions = builder.subscriptions;

        if (builder.brokerUrl == null) {
            brokerUrl = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
        } else {
            brokerUrl = builder.brokerUrl;
        }
    }

    /**
     * Configures the topics for the listeners registered in the {@link Builder#listenFor(org.app_chat.api.Listener)}
     * Starts the connection to the chat.
     */

    public void startChat() {
        connectionFactory = new ActiveMQConnectionFactory(brokerUrl);

        try {
            Connection connection = connectionFactory.createConnection();
            connection.start();

            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            for (final Listener listener : subscriptions) {

                String topicName = listener.getQueueName();

                Destination destination = session.createTopic(topicName);

                topicNamesToTopics.put(topicName, destination);

                MessageConsumer consumer = session.createConsumer(destination);

                MessageListener jmsListener = new MessageListener() {
                    @Override
                    public void onMessage(javax.jms.Message message) {
                        try {
                            if (message instanceof BytesMessage) {

                                BytesMessage bytesMessage = (BytesMessage) message;
                                byte[] messageBytes = new byte[(int) bytesMessage.getBodyLength()];
                                bytesMessage.readBytes(messageBytes);

                                listener.messageReceived(MessageUnMarshaller.getUnMarshaller(listener).unMarshall(messageBytes));
                            }
                        } catch (Exception e) {
                            throw Throwables.propagate(e);
                        }
                    }
                };

                consumer.setMessageListener(jmsListener);

            }
        } catch (JMSException e) {
            throw Throwables.propagate(e);
        }

        publisher = new PublisherImpl();
    }

    /**
     * Stops all of the connections for this client
     */

    public void stopChat() {
         //TODO Implement this
    }

    /**
     * Gets the publisher for this ChatRoom isntance
     * 
     * @return the Publisher
     */

    public Publisher getPublisher() {
        return publisher;
    }

    /**
     * Static method to return the {@link Builder} to configure and Construct this ChatRoom
     * 
     * @return the {@link Builder} to construct the ChatRoom
     */

    public static Builder builder() {
        return new Builder();
    }

    /**
     * The ActiveMQ implementation of the {@link Publisher}
     */

    public class PublisherImpl implements Publisher {

        /**
         * Method that publishes the message. If the topic for the message
         * does not already exist, a topic will be created with the default
         * queue name from the message
         * 
         * @param message the message to be published
         */
        @Override
        public void publish(Message message) {
            String messageDescriptor = message.getDescriptorForType().getFullName();
            byte[] marshalledMessage = message.toByteArray();
            MessageProducer producer;

            try {

                if (!(queueNamesToProducers.containsKey(messageDescriptor))) {
                    if (topicNamesToTopics.containsKey(messageDescriptor)) {
                        producer = session.createProducer(topicNamesToTopics.get(messageDescriptor));
                        queueNamesToProducers.put(messageDescriptor, producer);
                    } else {
                        Destination destination = session.createTopic(messageDescriptor);
                        topicNamesToTopics.put(messageDescriptor, destination);
                        producer = session.createProducer(destination);
                        queueNamesToProducers.put(messageDescriptor, producer);
                    }
                }

                BytesMessage bytesMessage = session.createBytesMessage();
                bytesMessage.writeBytes(marshalledMessage);

                queueNamesToProducers.get(messageDescriptor).send(bytesMessage);

            } catch (JMSException e) {
                throw Throwables.propagate(e);
            }
        }
    }

    /**
     * The Builder class that configures and constructs the ChatRoom
     */

    public static class Builder {

        private String brokerUrl;
        Set<Listener> subscriptions = Sets.newHashSet();

        /**
         * Set the URL to connect to the Broker Service
         * If this is {@literal null}, then the ChatRoom will use the
         * {@link ActiveMQConnectionFactory#DEFAULT_BROKER_URL}
         *  
         * @param brokerUrl the broker URL
         * @return  the {@link Builder} instance
         */

        public Builder setBrokerUrl(String brokerUrl) {
            this.brokerUrl = brokerUrl;
            return this;
        }

        /**
         * Add listeners. The listeners will automatically configure
         * the topics and queues needed for publishing and listening
         *
         * @param listener  the {@link Listener} to register
         * @return the {@link Builder} instance                                                   */

        public Builder listenFor(Listener listener) {
            this.subscriptions.add(listener);
            return this;
        }

        /**
         * Construct the configured chatroom instance
         *
         * @return  the {@link ChatRoom}
         */

        public ChatRoom build() {
            return new ChatRoom(this);
        }
    }
}
